package com.ww.microrpc.netty;

import com.ww.microrpc.client.Client;
import com.ww.microrpc.netty.common.RemotingHelper;
import com.ww.microrpc.netty.common.RemotingUtil;
import com.ww.microrpc.netty.exception.RemotingConnectException;
import com.ww.microrpc.netty.exception.RpcClientException;
import com.ww.microrpc.netty.handler.ProtocolFrameDecoderHandler;
import com.ww.microrpc.netty.handler.RpcDecoderHandler;
import com.ww.microrpc.netty.handler.RpcEncoderHandler;
import com.ww.microrpc.netty.handler.client.PaddingRequestContext;
import com.ww.microrpc.netty.handler.client.RpcClientHandler;
import com.ww.microrpc.netty.handler.client.RpcFuture;
import com.ww.microrpc.netty.protocol.ResponseMessage;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class NettyClient implements Client {
    private final static Logger logger = LoggerFactory.getLogger(NettyClient.class);

    private NettyClientConfig clientConfig;

    private Bootstrap bootstrap = new Bootstrap();

    /**
     * logging handler Sharable
     */
    private LoggingHandler loggingHandler = new LoggingHandler(LogLevel.DEBUG);

    private static final long LOCK_TIMEOUT_MILLIS = 3000;
    private final Lock lockChannelTables = new ReentrantLock();

    public NettyClient(NettyClientConfig clientConfig) {
        this.clientConfig = clientConfig;
        NettyCommon nettyCommon = new NettyCommon(clientConfig);
        EventLoopGroup eventLoopGroup = nettyCommon.buildEventLoopGroupSelector();
        //设置线程组
        this.bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        channel.pipeline().addLast(new ProtocolFrameDecoderHandler());
                        channel.pipeline().addLast(new RpcDecoderHandler());
                        channel.pipeline().addLast(new RpcEncoderHandler());
//                        channel.pipeline().addLast(new NettyConnectManageHandler());
                        channel.pipeline().addLast(loggingHandler);
                        // No-Sharable
                        channel.pipeline().addLast(new RpcClientHandler());
                    }
                })
                .option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(1024 * 10));

        if (this.clientConfig.getPort() != null && this.clientConfig.getPort() > 0) {
            bootstrap.localAddress(new InetSocketAddress(this.clientConfig.getIp(), this.clientConfig.getPort()));
        }

        // 设置发送区缓存
        if (this.clientConfig.getClientSocketSndBufSize() > 0) {
            logger.info("client set SO_SNDBUF to {}", this.clientConfig.getClientSocketSndBufSize());
            bootstrap.option(ChannelOption.SO_SNDBUF, this.clientConfig.getClientSocketSndBufSize());
        }

        // 设置接收区缓存
        if (this.clientConfig.getClientSocketRcvBufSize() > 0) {
            logger.info("client set SO_RCVBUF to {}",  this.clientConfig.getClientSocketRcvBufSize());
            bootstrap.option(ChannelOption.SO_RCVBUF, this.clientConfig.getClientSocketRcvBufSize());
        }

        //
        if (this.clientConfig.getWriteBufferLowWaterMark() > 0 && clientConfig.getWriteBufferHighWaterMark() > 0) {
            logger.info("client set netty WRITE_BUFFER_WATER_MARK to {},{}", this.clientConfig.getWriteBufferLowWaterMark(), this.clientConfig.getWriteBufferHighWaterMark());
            bootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(this.clientConfig.getWriteBufferLowWaterMark(), this.clientConfig.getWriteBufferHighWaterMark()));
        }
    }

    private ResponseMessage send(RequestHolder request) {
        return send(request, -1L, TimeUnit.SECONDS);
    }

    private ResponseMessage send(RequestHolder request, long timeout, TimeUnit unit) {
        RpcFuture rpcFuture = new RpcFuture(request);
        Channel channel = null;
        try {
            channel = this.getAndCreateChannel(request.getUrl());
            PaddingRequestContext.padding(request.getRequestId(), rpcFuture);
            ChannelFuture channelFuture = channel.writeAndFlush(request.getRequestMessage()).sync();
            if (!channelFuture.isSuccess()) {
                String msg = "send request failure, requestId: " + request.getRequestId();
                logger.error(msg);
                return ResponseMessage.failureResponse(request.getRequestId(), msg);
            }
            if (request.isAsync()) {
                return ResponseMessage.successResponse(request.getRequestId(), "send request success: " + request.getRequestId());
            }
            if (timeout > 0) {
                return rpcFuture.get(timeout, unit);
            } else {
                return rpcFuture.get();
            }
        } catch (RemotingConnectException | InterruptedException e) {
            logger.error("Send request exception: " + e.getMessage());
            throw new RpcClientException(e.getMessage(), e.getCause());
        }
    }


    private final ConcurrentMap<String /* addr */, NettyClient.ChannelWrapper> channelTables = new ConcurrentHashMap<String, NettyClient.ChannelWrapper>();

    private Channel getAndCreateChannel(final String addr) throws RemotingConnectException, InterruptedException {
        NettyClient.ChannelWrapper cw = this.channelTables.get(addr);
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }

        return this.createChannel(addr);
    }

    private Channel createChannel(final String addr) throws InterruptedException {
        NettyClient.ChannelWrapper cw = this.channelTables.get(addr);
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }

        if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
            try {
                boolean createNewConnection;
                cw = this.channelTables.get(addr);
                if (cw != null) {

                    if (cw.isOK()) {
                        return cw.getChannel();
                    } else if (!cw.getChannelFuture().isDone()) {
                        createNewConnection = false;
                    } else {
                        this.channelTables.remove(addr);
                        createNewConnection = true;
                    }
                } else {
                    createNewConnection = true;
                }

                if (createNewConnection) {
                    ChannelFuture channelFuture = this.bootstrap.connect(RemotingHelper.string2SocketAddress(addr));
                    logger.info("createChannel: begin to connect remote host[{}] asynchronously", addr);
                    cw = new NettyClient.ChannelWrapper(channelFuture);
                    this.channelTables.put(addr, cw);
                }
            } catch (Exception e) {
                logger.error("createChannel: create channel exception", e);
            } finally {
                this.lockChannelTables.unlock();
            }
        } else {
            logger.warn("createChannel: try to lock channel table, but timeout, {}ms", LOCK_TIMEOUT_MILLIS);
        }

        if (cw != null) {
            ChannelFuture channelFuture = cw.getChannelFuture();
            if (channelFuture.awaitUninterruptibly(this.clientConfig.getConnectTimeoutMillis())) {
                if (cw.isOK()) {
                    logger.info("createChannel: connect remote host[{}] success, {}", addr, channelFuture.toString());
                    return cw.getChannel();
                } else {
                    logger.warn("createChannel: connect remote host[" + addr + "] failed, " + channelFuture.toString(), channelFuture.cause());
                }
            } else {
                logger.warn("createChannel: connect remote host[{}] timeout {}ms, {}", addr, this.clientConfig.getConnectTimeoutMillis(),
                        channelFuture.toString());
            }
        }

        return null;
    }

    public void closeChannel(final Channel channel) {
        if (null == channel) {
            return;
        }
        try {
            if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
                    boolean removeItemFromTable = true;
                    NettyClient.ChannelWrapper prevCW = null;
                    String addrRemote = null;
                    for (Map.Entry<String, NettyClient.ChannelWrapper> entry : channelTables.entrySet()) {
                        String key = entry.getKey();
                        NettyClient.ChannelWrapper prev = entry.getValue();
                        if (prev.getChannel() != null) {
                            if (prev.getChannel() == channel) {
                                prevCW = prev;
                                addrRemote = key;
                                break;
                            }
                        }
                    }

                    if (null == prevCW) {
                        logger.info("eventCloseChannel: the channel[{}] has been removed from the channel table before", addrRemote);
                        removeItemFromTable = false;
                    }

                    if (removeItemFromTable) {
                        this.channelTables.remove(addrRemote);
                        logger.info("closeChannel: the channel[{}] was removed from channel table", addrRemote);
                        RemotingUtil.closeChannel(channel);
                    }
                } catch (Exception e) {
                    logger.error("closeChannel: close the channel exception", e);
                } finally {
                    this.lockChannelTables.unlock();
                }
            } else {
                logger.warn("closeChannel: try to lock channel table, but timeout, {}ms", LOCK_TIMEOUT_MILLIS);
            }
        } catch (InterruptedException e) {
            logger.error("closeChannel exception", e);
        }
    }

    public void closeChannel(final String addr, final Channel channel) {
        if (null == channel) {
            return;
        }
        final String addrRemote = null == addr ? RemotingHelper.parseChannelRemoteAddr(channel) : addr;

        try {
            if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
                    boolean removeItemFromTable = true;
                    final NettyClient.ChannelWrapper prevCW = this.channelTables.get(addrRemote);

                    logger.info("closeChannel: begin close the channel[{}] Found: {}", addrRemote, prevCW != null);

                    if (null == prevCW) {
                        logger.info("closeChannel: the channel[{}] has been removed from the channel table before", addrRemote);
                        removeItemFromTable = false;
                    } else if (prevCW.getChannel() != channel) {
                        logger.info("closeChannel: the channel[{}] has been closed before, and has been created again, nothing to do.",
                                addrRemote);
                        removeItemFromTable = false;
                    }

                    if (removeItemFromTable) {
                        this.channelTables.remove(addrRemote);
                        logger.info("closeChannel: the channel[{}] was removed from channel table", addrRemote);
                    }

                    RemotingUtil.closeChannel(channel);
                } catch (Exception e) {
                    logger.error("closeChannel: close the channel exception", e);
                } finally {
                    this.lockChannelTables.unlock();
                }
            } else {
                logger.warn("closeChannel: try to lock channel table, but timeout, {}ms", LOCK_TIMEOUT_MILLIS);
            }
        } catch (InterruptedException e) {
            logger.error("closeChannel exception", e);
        }
    }

    @Override
    public ResponseMessage execute(RequestHolder request) throws IOException {
        return this.send(request);
    }

    static class ChannelWrapper {
        private final ChannelFuture channelFuture;

        public ChannelWrapper(ChannelFuture channelFuture) {
            this.channelFuture = channelFuture;
        }

        public boolean isOK() {
            return this.channelFuture.channel() != null && this.channelFuture.channel().isActive();
        }

        public boolean isWritable() {
            return this.channelFuture.channel().isWritable();
        }

        private Channel getChannel() {
            return this.channelFuture.channel();
        }

        public ChannelFuture getChannelFuture() {
            return channelFuture;
        }
    }

    class NettyConnectManageHandler extends ChannelDuplexHandler {
        @Override
        public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
                            ChannelPromise promise) throws Exception {
            final String local = localAddress == null ? "UNKNOWN" : RemotingHelper.parseSocketAddressAddr(localAddress);
            final String remote = remoteAddress == null ? "UNKNOWN" : RemotingHelper.parseSocketAddressAddr(remoteAddress);
            logger.info("NETTY CLIENT PIPELINE: CONNECT  {} => {}", local, remote);

            super.connect(ctx, remoteAddress, localAddress, promise);
        }

        @Override
        public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            logger.info("NETTY CLIENT PIPELINE: DISCONNECT {}", remoteAddress);
            closeChannel(ctx.channel());
            super.disconnect(ctx, promise);
        }

        @Override
        public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            logger.info("NETTY CLIENT PIPELINE: CLOSE {}", remoteAddress);
            closeChannel(ctx.channel());
            super.close(ctx, promise);
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state().equals(IdleState.ALL_IDLE)) {
                    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                    logger.warn("NETTY CLIENT PIPELINE: IDLE exception [{}]", remoteAddress);
                    closeChannel(ctx.channel());
                }
            }

            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            logger.warn("NETTY CLIENT PIPELINE: exceptionCaught {}", remoteAddress);
            logger.warn("NETTY CLIENT PIPELINE: exceptionCaught exception.", cause);
            closeChannel(ctx.channel());
        }
    }
}
